package cn.iocoder.yudao.module.ai.graph.controller;

import cn.iocoder.yudao.module.ai.graph.controller.GraphProcess.GraphProcess;
import com.alibaba.cloud.ai.graph.*;
import com.alibaba.cloud.ai.graph.checkpoint.config.SaverConfig;
import com.alibaba.cloud.ai.graph.checkpoint.constant.SaverEnum;
import com.alibaba.cloud.ai.graph.checkpoint.savers.MemorySaver;
import com.alibaba.cloud.ai.graph.exception.GraphRunnerException;
import com.alibaba.cloud.ai.graph.exception.GraphStateException;
import com.alibaba.cloud.ai.graph.state.StateSnapshot;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import java.util.HashMap;
import java.util.Map;

/**
 * 客服工作流控制器 - 演示完整的人工打断和恢复功能
 * 
 * API 接口：
 * 1. POST /ai/graph/customer-service/start - 启动新的客服会话
 * 2. POST /ai/graph/customer-service/review - 人工审核并恢复流程
 * 3. GET /ai/graph/customer-service/state/{threadId} - 查看当前状态
 * 4. GET /ai/graph/customer-service/history/{threadId} - 查看历史记录
 * 
 * @author yudao
 * @since 2025/10/17
 */
@RestController
@RequestMapping("/ai/graph/customer-service")
public class CustomerServiceController {

    private static final Logger logger = LoggerFactory.getLogger(CustomerServiceController.class);

    private final CompiledGraph compiledGraph;

    @Autowired
    public CustomerServiceController(@Qualifier("customerServiceGraph") StateGraph stateGraph) 
            throws GraphStateException {
        // 配置内存状态保存器（实际项目中可以使用 Redis 等持久化方案）
        SaverConfig saverConfig = SaverConfig.builder()
                .register(SaverEnum.MEMORY.getValue(), new MemorySaver())
                .build();
        
        // 编译工作流，配置在 human_review 节点前打断
        this.compiledGraph = stateGraph.compile(
            CompileConfig.builder()
                .saverConfig(saverConfig)
                .interruptBefore("human_review")  // 在人工审核前打断
                // .interruptBefore("satisfaction_survey")  // 可选：在满意度调查前也打断
                .build()
        );
        
        logger.info("✓ 客服工作流已初始化，打断点：human_review");
    }

    /**
     * 启动新的客服会话
     * 
     * @param request 客户请求
     * @return SSE 流式响应
     */
    @PostMapping(value = "/start", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> startCustomerService(@RequestBody CustomerServiceRequest request) 
            throws GraphRunnerException {
        
        logger.info("\n" + "=".repeat(80));
        logger.info("🎯 启动客服工作流");
        logger.info("客户 ID：{}", request.getCustomerId());
        logger.info("客户问题：{}", request.getQuery());
        logger.info("线程 ID：{}", request.getThreadId());
        logger.info("=".repeat(80));
        
        // 配置运行参数
        RunnableConfig runnableConfig = RunnableConfig.builder()
                .threadId(request.getThreadId())
                .build();
        
        // 准备初始状态
        Map<String, Object> initialState = new HashMap<>();
        initialState.put("customer_id", request.getCustomerId());
        initialState.put("customer_query", request.getQuery());
        initialState.put("customer_context", request.getContext());
        initialState.put("thread_id", request.getThreadId());
        
        // 创建流处理器
        GraphProcess graphProcess = new GraphProcess(this.compiledGraph);
        Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().unicast().onBackpressureBuffer();
        
        // 启动工作流（流式输出）
        Flux<NodeOutput> nodeOutputFlux = compiledGraph.fluxStream(initialState, runnableConfig);
        graphProcess.processStream(nodeOutputFlux, sink);
        
        return sink.asFlux()
                .doOnCancel(() -> logger.info("❌ 客户端断开连接"))
                .doOnComplete(() -> logger.info("✓ 工作流已暂停，等待人工审核"))
                .doOnError(e -> logger.error("✗ 工作流执行错误", e));
    }

    /**
     * 人工审核并恢复工作流
     * 
     * @param request 审核决策
     * @return SSE 流式响应
     */
    @PostMapping(value = "/review", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> reviewAndResume(@RequestBody ReviewRequest request) 
            throws GraphRunnerException {
        
        logger.info("\n" + "=".repeat(80));
        logger.info("👤 人工审核并恢复工作流");
        logger.info("线程 ID：{}", request.getThreadId());
        logger.info("审核结果：{}", request.isApproved() ? "✓ 通过" : "✗ 不通过");
        logger.info("决策：{}", request.getDecision());
        logger.info("=".repeat(80));
        
        // 配置运行参数
        RunnableConfig runnableConfig = RunnableConfig.builder()
                .threadId(request.getThreadId())
                .build();
        
        // 获取当前状态快照
        StateSnapshot stateSnapshot = this.compiledGraph.getState(runnableConfig);
        OverAllState state = stateSnapshot.state();
        state.withResume(); // 标记为恢复模式
        
        // 准备人工反馈数据
        Map<String, Object> feedbackData = new HashMap<>();
        feedbackData.put("approved", request.isApproved());
        feedbackData.put("modified_reply", request.getModifiedReply());
        feedbackData.put("decision", request.getDecision());
        feedbackData.put("comments", request.getComments());
        
        // 设置人工反馈
        state.withHumanFeedback(new OverAllState.HumanFeedback(feedbackData, ""));
        
        // 创建流处理器
        GraphProcess graphProcess = new GraphProcess(this.compiledGraph);
        Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().unicast().onBackpressureBuffer();
        
        // 从打断点恢复执行
        Flux<NodeOutput> nodeOutputFlux = compiledGraph.fluxStreamFromInitialNode(state, runnableConfig);
        graphProcess.processStream(nodeOutputFlux, sink);
        
        return sink.asFlux()
                .doOnCancel(() -> logger.info("❌ 客户端断开连接"))
                .doOnComplete(() -> logger.info("✓ 工作流执行完成"))
                .doOnError(e -> logger.error("✗ 工作流执行错误", e));
    }

    /**
     * 查看当前工作流状态
     * 
     * @param threadId 线程 ID
     * @return 状态信息
     */
    @GetMapping("/state/{threadId}")
    public Map<String, Object> getState(@PathVariable String threadId) throws GraphRunnerException {
        RunnableConfig runnableConfig = RunnableConfig.builder()
                .threadId(threadId)
                .build();
        
        StateSnapshot stateSnapshot = this.compiledGraph.getState(runnableConfig);
        
        Map<String, Object> result = new HashMap<>();
        result.put("thread_id", threadId);
        result.put("next_node", stateSnapshot.next());
        result.put("state_data", stateSnapshot.state().data());
        
        logger.info("📊 查询状态 - 线程：{}, 下一节点：{}", threadId, stateSnapshot.next());
        
        return result;
    }

    /**
     * 查看工作流历史记录
     * 
     * @param threadId 线程 ID
     * @return 历史记录列表
     */
    @GetMapping("/history/{threadId}")
    public Map<String, Object> getHistory(@PathVariable String threadId) throws GraphRunnerException {
        RunnableConfig runnableConfig = RunnableConfig.builder()
                .threadId(threadId)
                .build();
        
        // 获取历史状态快照
        java.util.Collection<StateSnapshot> historyCollection = this.compiledGraph.getStateHistory(runnableConfig);
        java.util.List<StateSnapshot> history = new java.util.ArrayList<>(historyCollection);
        
        Map<String, Object> result = new HashMap<>();
        result.put("thread_id", threadId);
        result.put("total_checkpoints", history.size());
        
        java.util.List<Map<String, Object>> historyData = new java.util.ArrayList<>();
        for (StateSnapshot snapshot : history) {
            Map<String, Object> item = new HashMap<>();
            item.put("next_node", snapshot.next());
            item.put("state", snapshot.state().data());
            historyData.add(item);
        }
        result.put("history", historyData);
        
        logger.info("📜 查询历史 - 线程：{}, 共 {} 个检查点", threadId, history.size());
        
        return result;
    }
}

/**
 * 客户服务请求
 */
@lombok.Data
class CustomerServiceRequest {
    private String customerId;
    private String query;
    private String context = "";
    private String threadId;
}

/**
 * 审核请求
 */
@lombok.Data
class ReviewRequest {
    private String threadId;
    private boolean approved;
    private String modifiedReply = "";
    private String decision = "send_reply"; // send_reply | transfer_to_human | escalate | end
    private String comments = "";
}

